package com.test.task;

import java.util.LinkedList;
import java.util.List;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

/**
 * 流水线中的工作类，为一个抽象类，定义了所有的接口，包含了一部分实现
 * 
 * @author ws
 * 
 */
public abstract class Task<T> implements Runnable {
    /**
     * 默认的工作线程类的大小
     */
    public static final int DEFAULT_WORKER_ACCOUNT = 20;

    /**
     * 打印日志，使用Log4j，对Log4j作统一的级别管理
     */
    private final static Logger log = LogManager.getLogger(Task.class);

    /**
     * 工作流水是否已经找开的标志
     */
    private boolean isOpen = false;

    /**
     * 消息队列的最大值
     */
    private int maxQueueSize = 10000;

    /**
     * 消息块的队列
     */
    private List<T> msgQueue = new LinkedList<T>();

    /**
     * 工作流水的名称
     */
    private String name = this.getClass().getName();

    /**
     * 下一级流水，如果不为空，则在work函数执行完毕以后，在putNext函数里，将消息传递给下一级流水
     */
    private Task nextTask = null;

    /**
     * 消息队列的锁，涉及到消息队列的操作时，需要获取此锁，以作线程同步
     */
    private Object queueLock = new Object();

    /**
     * 需要被停止的线程的大小
     */
    private int reduceAccount = 0;

    private ThreadGroup threadGroup = new ThreadGroup(name);

    /**
     * 工作线程的锁
     */
    private Object workerLock = new Object();

    /**
     * 将一个新的数据加入到工作队列中，将同时唤醒所有的工作线程，如果此时队列中消息的大小已经达到上限，则需要进行等待
     * 
     * @param t
     */
    @SuppressWarnings("unchecked")
    public void addMsg(Object obj) {
	T t = (T) obj;

	synchronized (this.queueLock) {
	    // 检查队列中的消息数量是否超标
	    while (this.msgQueue.size() >= this.maxQueueSize && this.isOpen) {
		try {
		    this.queueLock.wait();
		} catch (InterruptedException e) {
		    log.error("", e);
		}
	    }

	    this.msgQueue.add(t);

	    // 唤醒所有的工作线程
	    this.notifyAllWorker();
	}
    }

    /**
     * 增加工作线程
     * 
     * @param num
     */
    public void addWorker(int num) {
	for (int it = 0; it < num; it++) {
	    initWorker();
	}
    }

    /**
     * 停止所有的工作线程，并将open标志置为空
     * 
     */
    public void finish() {
	this.isOpen = false;
	this.reduceWorker(this.threadGroup.activeCount());
    }

    /**
     * 返回当前处于活动状态的工作线程的数量
     * 
     * @return
     */
    public int getActiveAccount() {
	return this.threadGroup.activeCount();
    }

    public int getMaxQueueSize() {
	return maxQueueSize;
    }

    /**
     * 获取当前消息队列中的第一个消息块
     * 
     * @return
     */
    public T getMsgBlock() {
	synchronized (this.queueLock) {
	    if (!this.msgQueue.isEmpty()) {
		return this.msgQueue.remove(0);
	    } else {
		return null;
	    }
	}
    }

    public String getName() {
	return name;
    }

    public Object getQueueLock() {
	return queueLock;
    }

    public ThreadGroup getThreadGroup() {
	return threadGroup;
    }

    /**
     * 初始化一个工作线程类，同时将流水是否开启标志置为true
     * 
     */
    public void initWorker() {
	synchronized (workerLock) {
	    Thread thread = new Thread(this.threadGroup, this, this.threadGroup
		    .getName());
	    thread.start();
	}
    }

    /**
     * 唤醒所有的工作线程
     * 
     */
    public void notifyAllWorker() {
	synchronized (this.queueLock) {
	    this.queueLock.notifyAll();
	}
    }

    /**
	 * 
	 *
	 */
    public void open() {
	open(DEFAULT_WORKER_ACCOUNT);
    }

    public void open(int workerAccount) {
	if (!this.isOpen) {
	    isOpen = true;
	    addWorker(workerAccount);
	}
    }

    public void putNextTask(Object obj) {
	if (this.nextTask != null) {
	    this.nextTask.addMsg(obj);
	}
    }

    /**
     * 将要停止的线程数自减１，同时唤醒所有的工作线程
     * 
     */
    public void reduceWorker() {
	synchronized (workerLock) {
	    if (this.threadGroup.activeCount() > 0) {
		this.reduceAccount++;
	    }

	    this.notifyAllWorker();
	}
    }

    /**
     * 停止指定数目的线程数
     * 
     */
    public void reduceWorker(int num) {
	for (int it = 0; it < num; it++) {
	    reduceWorker();
	}
    }

    /**
     * 线程的工作方法，主要工作就是从队列中取出数据，并调用工作方法
     */
    public void run() {
	log.info("工作流水[" + this.name + "]工作线程启动，当前活动线程为"
		+ this.threadGroup.activeCount());
	while (true) {
	    // 检查是否收到有减少流水工作线程的信号
	    if (this.reduceAccount > 0) {
		synchronized (workerLock) {
		    this.reduceAccount--;
		    break;
		}
	    }

	    // 从工作队列中获取消息块，并调用工作方法
	    T msg = null;
	    synchronized (this.queueLock) {
		if (this.msgQueue.isEmpty()) {
		    try {
			this.queueLock.wait();
			continue;
		    } catch (InterruptedException e) {
			log.error("工作流水[" + this.name + "]工作线程意外退出", e);
			break;
		    }
		} else {
		    msg = this.getMsgBlock();
		}
	    }

	    // 调用工作类对消息块进行处理
	    try {
		this.work(msg);
	    } catch (Exception e) {
		log.error("Work thread throws Exception: ", e);
	    }
	}

	log.info("工作流水[" + this.name + "]工作线程将减少1，当前活动线程为"
		+ (this.threadGroup.activeCount() - 1));
    }

    public void setMaxQueueSize(int maxQueueSize) {
	this.maxQueueSize = maxQueueSize;
    }

    public void setName(String name) {
	this.name = name;
    }

    public void setQueueLock(Object queueLock) {
	this.queueLock = queueLock;
    }

    public void setThreadGroup(ThreadGroup threadGroup) {
	this.threadGroup = threadGroup;
    }

    /**
     * 设置工作线程的数量，必须在本实例open之后，设置才会生效
     * 
     * @param num
     */
    public void setWorkerAccount(int num) {
	if (this.isOpen) {
	    if (num - this.threadGroup.activeCount() > 0) {
		this.addWorker(num - this.threadGroup.activeCount());
	    }
	}
    }

    /**
     * 流水的工作类
     * 
     */
    protected abstract void work(T t);

}
